package com.taimanetworks.kafka;

import com.taimanetworks.kafka.parseTsp.GpsData;
import com.taimanetworks.utils.LngLatConvertUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisPoolConfig;

import java.util.HashSet;
import java.util.Set;

public class RedisSinkFunction extends RichSinkFunction<GpsData> {

    private  static JedisCluster jedisCluster =null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Set<HostAndPort> nodes = new HashSet<>();

        nodes.add(new HostAndPort("172.22.2.112", 6379));
        nodes.add(new HostAndPort("172.22.2.112", 6380));
        nodes.add(new HostAndPort("172.22.2.113", 6379));
        nodes.add(new HostAndPort("172.22.2.113", 6380));
        nodes.add(new HostAndPort("172.22.2.114", 6379));
        nodes.add(new HostAndPort("172.22.2.114", 6380));
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();

        // 最大空闲数
        jedisPoolConfig.setMaxIdle(10);
        // 最大连接数
        jedisPoolConfig.setMaxTotal(30);
        jedisPoolConfig.setMinIdle(1);
        // 最大允许等待时间，如果超过这个时间还未获取到连接，则会报JedisException异常：
        jedisPoolConfig.setMaxWaitMillis(3000);
        // 在还会给pool时，是否提前进行validate操作
        jedisPoolConfig.setTestOnBorrow(true);

        jedisCluster = new JedisCluster(nodes, 1000, 2000, 3000, "Tima@!QAZ", "cluster", jedisPoolConfig);


    }

    @Override
    public void invoke(GpsData gpsData, Context context) throws Exception {
        String longitude_str = gpsData.getLongitude();
        String latitude_str = gpsData.getLatitude();
        String longituders = longitude_str.substring(3, longitude_str.length());
        String latituders = latitude_str.substring(3, latitude_str.length());

        double[] resultlist = LngLatConvertUtil.transform(Double.valueOf(longituders),Double.valueOf(latituders));

        String result=gpsData.getMilleage()+"&&"+gpsData.getSoc()+"&&"+String.valueOf(resultlist[0])+"&&"+String.valueOf(resultlist[1])+"&&"+gpsData.getGps_time();

        jedisCluster.hset("car_signal",gpsData.getVin(),result);
    }

    @Override
    public void close() throws Exception {
        super.close();
        jedisCluster.close();
    }

}
